/*
 * @ProjectName: 综合安防
 * @Copyright:   2018 HangZhou Hikvision System Technology Co., Ltd. All Right Reserved.
 * @address:     http://www.hikvision.com
 * @date:        2018年01月29日 14:21
 * @description: 本内容仅限于杭州海康威视系统技术公有限司内部使用，禁止转发.
 */
package com.capsule.chapter.queue;


import org.apache.commons.collections4.CollectionUtils;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * <p>
 * 被观察者，也就是消息队列
 * 实现了Observerable接口，对Observerable接口的三个方法进行了具体实现
 * </p>
 *
 * @author xiachaoyang
 * @version V1.0
 * @date 2018年07月04日 10:02
 * @modificationHistory=========================逻辑或功能性重大变更记录
 * @modify By: {修改人} 2018年07月04日
 * @modify reason: {方法名}:{原因}
 * ...
 */
public class MsgQueue extends LinkedBlockingQueue<Object> implements Observerable {

    private static Integer QUEUE_SIZE = 1000;

    /**
     * 静态内部类式单例
     */
    private static class SingletonHolder {
        private static final MsgQueue instance = new MsgQueue();
    }

    private MsgQueue() {
        super(QUEUE_SIZE);
        list = new ArrayList<Observer>();
    }

    public static final MsgQueue getInstance() {
        return MsgQueue.SingletonHolder.instance;
    }

    /**
     * 注意到这个List集合的泛型参数为Observer接口，设计原则：面向接口编程而不是面向实现编程 观察者集合，以便通知观察者
     */
    private List<Observer> list;

    private String message;

    private boolean isClose = false;

    /**
     * 注册
     *
     * @param observer
     */
    @Override
    public void registerObserver(Observer observer) {
        list.add(observer);
    }

    /**
     * 移除
     *
     * @param observer
     */
    @Override
    public void removeObserver(Observer observer) {
        if (!list.isEmpty())
            list.remove(observer);
    }

    /**
     * 通知
     */
    @Override
    public void notifyObserver() throws InterruptedException {
        if (CollectionUtils.isEmpty(list)) {
            return;
        }
        for (int i = 0; i < list.size(); i++) {
            Observer oserver = list.get(i);
            //队列中无数据时会发生阻塞
            oserver.update(this.take());
        }
    }

    /**
     * 停止队列监听
     */
    public void stop() {
        if (CollectionUtils.isEmpty(list)) {
            return;
        } else {
            close();
            System.out.println("停止对队列变化进行监听....消息继续入队但是不处理");
        }
    }

    /**
     * 激活并发起通知
     * @throws InterruptedException
     */
    public void active() throws InterruptedException {
        isClose = false;
        System.out.println("重新激活线程：数目："+this.size());
        notifyObserver();
    }

    public void close() {
        isClose = true;
    }

    /**
     * 消息推送
     *
     * @param msg
     */
    public void publishMsg(String msg) throws InterruptedException {
        this.message = msg;
        this.put(msg);
        //消息更新，通知所有观察者
        if (!isClose) {
            notifyObserver();
        }

    }
}
